package cn.jly.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author lanyangji
 * @date 2019/12/2 22:14
 */
object SparkStreaming01_PortWordCount {

  def main(args: Array[String]): Unit = {

    // 使用netcat工具向9999端口不断发送数据，通过sparkStreaming读取端口数据并统计不同单词出现的次数
    // 初始化配置信息
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming01_PortWordCount")
    // 初始化 spark的 StreamingContext对象
    // 将 4 秒内得到的数据当成一个整体，向下执行
    val streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(4))

    // 从指定的端口中获取数据
    // 通过检测端口，创建 DStream, 读进来的数据为一行一行
    val lineStreams: ReceiverInputDStream[String] = streamingContext.socketTextStream("hadoop102", 9999)

    // 将每一行数据进行切分，形成一个个单词
    val wordStreams: DStream[String] = lineStreams.flatMap(_.split(" "))

    // 将单词映射成元组
    val wordAndOneStreams: DStream[(String, Int)] = wordStreams.map((_, 1))

    // 将相同的单词进行统计
    val wordAndCountStreams: DStream[(String, Int)] = wordAndOneStreams.reduceByKey(_ + _)

    // 打印
    wordAndCountStreams.print()
    wordAndCountStreams.saveAsTextFiles("out_11", ".wordCount")

    // 不能停止采集程序
    // 启动 spark的 StreamingContext
    streamingContext.start()
    streamingContext.awaitTermination()

    // 注意，为了方便测试查看结果，将log4j配置文件放在类路径下，日志打印级别改为 WARN，即只打印错误日志
  }
}
